import { Component, OnInit } from '@angular/core';
import { Observable, Subject, asapScheduler, pipe, of, from, interval, merge, fromEvent, timer, observable, Subscriber, ConnectableObservable, range, BehaviorSubject } from 'rxjs';
import { map, filter, scan, take, concatAll, concatMap, takeUntil, retry, retryWhen, mapTo, multicast, publish, refCount, share, publishReplay, publishBehavior, publishLast, tap, takeLast, withLatestFrom } from 'rxjs/operators';
import { webSocket } from 'rxjs/webSocket';
import { ajax } from 'rxjs/ajax';
import { TestScheduler } from 'rxjs/testing';
import { asap } from 'rxjs/internal/scheduler/asap';
import { animationFrame } from 'rxjs/internal/scheduler/animationFrame';
import { async } from 'rxjs/internal/scheduler/async';
import { queue } from '../../node_modules/rxjs/internal/scheduler/queue';

@Component({
    selector: 'app-multicast',
    template: `<input /> <button>output</button>`,
    // styleUrls: ['./name.component.scss']
})
export class MulticastComponent implements OnInit {
    constructor() { }

    ngOnInit(): void {
        // 冷流
        // const cold$ = interval(1000);
        // cold$.subscribe(x => console.log('cold sub 1:' + x));
        // setTimeout(() => {
        //     cold$.subscribe(x => console.log('cold sub 2:' + x));
        // }, 2000);

        // 冷流 => 热流
        // const cold2$ = interval(1000);
        // const hot2$ = new Subject();
        // cold2$.subscribe(hot2$); // 等价于cold2$.subscribe((data) => { hot2$.next(data) });
        // hot2$.subscribe((data) => { console.log('subject sub 1:' + data) });
        // setTimeout(() => {
        //     hot2$.subscribe(data => console.log('subject sub 2:' + data));
        // }, 2000);

        // Subject
        // const subject = new Subject();
        // const subscription1 = subject.subscribe(
        //     value => console.log('observer 1 get value:' + value),
        //     error => console.log('observer 1 get error:' + error),
        //     () => console.log('observer 1 complete')
        // );
        // subject.next(1);
        // const subscription2 = subject.subscribe(
        //     value => console.log('observer 2 get value:' + value),
        //     error => console.log('observer 2 get error:' + error),
        //     () => console.log('observer 2 complete')
        // );
        // subject.next(2)
        // subscription1.unsubscribe();
        // subject.complete();

        // Subject - 多上游数据源 1 
        // const tick1$ = interval(1000).pipe(mapTo('a'), take(2));
        // const tick2$ = interval(1000).pipe(mapTo('b'), take(2));
        // const sub = new Subject();

        // tick1$.subscribe(sub);
        // tick2$.subscribe(sub);
        // sub.subscribe(value => console.log('observer 1:' + value));
        // sub.subscribe(value => console.log('observer 2:' + value));

        // Subject - 多上游数据源 2 
        // const tick1$ = interval(1000).pipe(mapTo('a'), take(2));
        // const tick2$ = interval(1000).pipe(mapTo('b'), take(2));
        // const sub = new Subject();

        // const tick3$ = merge(tick1$, tick2$);
        // tick3$.subscribe(sub);
        // sub.subscribe(value => console.log('observer 1:' + value));
        // sub.subscribe(value => console.log('observer 2:' + value));

        // 操作符
        // const cold$ = interval(1000).pipe(take(3));
        // const hot$ = cold$.pipe(multicast(new Subject())) as ConnectableObservable<any>;
        // hot$.connect();
        // hot$.subscribe(x => console.log('observer 1:' + x));

        // multicast
        // const cold$ = interval(1000).pipe(take(3));
        // const hot$ = cold$.pipe(multicast(new Subject())) as ConnectableObservable<any>;
        // hot$.connect();
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 1500);
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 5000);

        // refCount
        // const cold$ = interval(1000).pipe(take(3));
        // const hot$ = cold$.pipe(multicast(new Subject()), refCount()) as ConnectableObservable<any>;
        // hot$.subscribe(x => console.log('observer 1:' + x), error => console.log(error), () => console.log('complete'));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x), error => console.log(error), () => console.log('complete'));
        // }, 1500);
        // // 由于subject 已结束 因此ob 3无法得到任何数据
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 5000);


        // refCount + factory = share
        // const cold$ = interval(1000).pipe(take(3));
        // const subjectFactory = () => { return new Subject() };
        // const hot$ = cold$.pipe(multicast(subjectFactory), refCount()) as ConnectableObservable<any>;
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 1500);
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 5000);

        // publish
        // const cold$ = interval(1000).pipe(take(3));
        // const subjectFactory = () => { return new Subject() };
        // const hot$ = cold$.pipe(publish(), refCount()) as ConnectableObservable<any>;
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 1500);

        // share = publish + refcount + replay
        // const cold$ = interval(1000).pipe(take(3));
        // const hot$ = cold$.pipe(share()) as ConnectableObservable<any>;
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 1500);
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 5000);

        // async subject
        // const cold$ = interval(1000).pipe(take(5));
        // const hot$ = cold$.pipe(publishLast(), refCount()) as ConnectableObservable<any>;
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 2500);
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 7000);

        // replay subject
        // const cold$ = interval(1000).pipe(take(5));
        // const hot$ = cold$.pipe(publishReplay(), refCount()) as ConnectableObservable<any>;  //注意一般需要设置参数-即缓冲区大小，防止内存泄漏
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 2500);
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 6000);

        // behavior subject
        // const cold$ = interval(1000).pipe(take(5));
        // const hot$ = cold$.pipe(publishBehavior(-1), refCount()) as ConnectableObservable<any>;
        // hot$.subscribe(x => console.log('observer 1:' + x));
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 2:' + x));
        // }, 2500);
        // setTimeout(() => {
        //     hot$.subscribe(x => console.log('observer 3:' + x));
        // }, 7000);

        // Scheduler 引子
        // const source3$ = interval(1000);
        // console.log('before subscribe 3');
        // source3$.subscribe(
        //     value => console.log('sub3:' + value),
        //     error => console.error,
        //     () => console.log('complete 3')
        // );
        // console.log('after subscribe 3');

        // Scheduler
        // const source$ = range(1, 5);
        // console.log('before subscribe');
        // source$.subscribe(
        //     value => console.log('sub:' + value),
        //     error => console.error,
        //     () => console.log('complete')
        // );
        // console.log('after subscribe');

        // ASAP Scheduler
        // const source2$ = range(1, 5, asap);
        // console.log('before subscribe 2');
        // source2$.subscribe(
        //     value => console.log('sub2:' + value),
        //     error => console.error,
        //     () => console.log('complete 2')
        // );
        // console.log('after subscribe 2');

        // const source3$ = interval(1000, queue);
        // console.log('before subscribe 3');
        // source3$.subscribe(
        //     value => console.log('sub3:' + value),
        //     error => console.error,
        //     () => console.log('complete 3')
        // );
        // console.log('after subscribe 3');

        // ASAP/ASYNC/QUEUE Scheduler
        // console.log('before scheduler');
        // async.schedule(() => console.log('async'));
        // asap.schedule(() => console.log('asap'));
        // queue.schedule(() => console.log('queue'));
        // console.log('end scheduler');

        // const source$ = interval(1000);
        // const source2$ = source$.pipe(tap(value => console.log(value)));
        // source2$.subscribe(
        //     value => console.log('sub:' + value),
        //     error => console.error,
        //     () => console.log('complete')
        // );


        /// Test Scheduler
        const sb1$ = new BehaviorSubject(-1);
        const sb2$ = new BehaviorSubject(-2);
        // const cold$ = interval(1000).pipe(take(10));
        // const cold2$ = interval(1500).pipe(take(10));
        // cold$.subscribe(sb1$);
        // cold2$.subscribe(sb2$);
        // const sb1 =  cold$.pipe(publishBehavior(-1)) as ConnectableObservable<any>;
        // const sb2 =  cold2$.pipe(publishBehavior(-2)) as ConnectableObservable<any>;
        // sb1$.connect();
        // sb2$.connect();
        sb1$.pipe(withLatestFrom(sb2$)).subscribe(x => {
            console.log('observable 1000:' + x[0]);
            console.log('observable 1500:' + x[1]);
        });

        const input$ = fromEvent(document.querySelector('input'), 'input');
        const stop$ = fromEvent(document.querySelector('button'), 'click');
        // input$.pipe(takeUntil(stop$), map(event => (<any>event.target).value)).subscribe(value => console.log(value));

        // stop$.subscribe(value => console.log(document.querySelector('input').value));

        merge(input$, stop$).pipe(
            map(x => (<any>event.target).value),
            scan((acc, curr) => {
                if (curr === '') {
                    return [0, acc];
                } else {
                    return acc = curr;
                }
            }))
            .subscribe(value => {
                if (value instanceof Array) {
                    console.log(value[1]);
                }
            });
    }
}
